package com.atguigu.realtime.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV1;
import com.atguigu.realtime.bean.TableProcess;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.FlinkSinkUtil;
import com.atguigu.realtime.util.JdbcUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import lombok.val;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;


/**
 * @Author lzc
 * @Date 2022/5/17 13:50
 */
public class DimApp extends BaseAppV1 {
    public static void main(String[] args) {
        new DimApp().init(2001, 2, "DimApp", Constant.TOPIC_ODS_DB);
    }
    
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          DataStreamSource<String> stream) {
        // 1. 先消费 ods_db数据
        SingleOutputStreamOperator<String> eltedDataStream = etlData(stream);
        // 2. 通过cdc读取 配置表数据
        SingleOutputStreamOperator<TableProcess> tpStream = readTableProcess(env);
//        tpStream.print();
       // 3. 把数据流和配置流进行connect
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dataStream = connect(eltedDataStream, tpStream);
        // 4. 过滤掉不需要的字段
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream = filterNoColumns(dataStream);
        // 5. 把数据写入到HBase(phoenix)中
        writeToPhoenix(resultStream);
        
        
    }
    
    private void writeToPhoenix(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> stream) {
        /*
            能不能使用jdbc sink
            
             不能, 因为要写入的数据包括多个表
        
         */
        stream.addSink(FlinkSinkUtil.getPhoenixSink());
        
    }
    
    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> filterNoColumns(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dataTpStream) {
      return  dataTpStream.map(new MapFunction<Tuple2<JSONObject, TableProcess>, Tuple2<JSONObject, TableProcess>>() {
            @Override
            public Tuple2<JSONObject, TableProcess> map(Tuple2<JSONObject, TableProcess> t) throws Exception {
                JSONObject data = t.f0;
                List<String> cs = Arrays.asList(t.f1.getSink_columns().split(","));
    
                data.keySet().removeIf(key -> !cs.contains(key));
                return t;
            }
        });
    }
    
    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connect(SingleOutputStreamOperator<String> dataStream,
                                                                                 SingleOutputStreamOperator<TableProcess> tpStream) {
      
        // 1. 把数据流中的数据解析成jsonObject
        SingleOutputStreamOperator<JSONObject> dataJsonStream = dataStream.map(JSON::parseObject);
        
        // 2. 把配置做成广播流
        // 广播状态的本质是一个map
        // key  表名
        // value TableProcess对象
        MapStateDescriptor<String, TableProcess> tpStateDesc = new MapStateDescriptor<>("tpState", String.class, TableProcess.class);
        BroadcastStream<TableProcess> tpBcStream = tpStream.broadcast(tpStateDesc);
        // 3. 数据流和广播流进行connect
       return dataJsonStream
            .connect(tpBcStream)
            .process(new BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject, TableProcess>>() {
                
                
                private Connection conn;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    // 一个并行度一个连接
                    String driver = Constant.PHOENIX_DRIVER;
                    String url = Constant.PHOENIX_URL;
                    conn = JdbcUtil.getPhoenixConnection(driver, url);
                }
                
                @Override
                public void close() throws Exception {
                    if (conn != null && !conn.isClosed()) {
                        conn.close();
                    }
                }
                
                @Override
                public void processElement(JSONObject obj,
                                           ReadOnlyContext ctx,
                                           Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                    // 1. 从广播状态中读取这个条数据对应的配置信息
                    ReadOnlyBroadcastState<String, TableProcess> tpState = ctx.getBroadcastState(tpStateDesc);
                    
                    // 2. 把数据和配置信息组成一个元组, 放入都后序流中
                    String table = obj.getString("table");
                    TableProcess tp = tpState.get(table);
                    // ods_db的数据很多, 有维度, 也有实时
                    if (tp != null) {
                        out.collect(Tuple2.of(obj.getJSONObject("data"), tp));
                    }
                }
                
                //
                @Override
                public void processBroadcastElement(TableProcess tp,
                                                    Context ctx,
                                                    Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                    
                    // 处理广播流中的元素
                    // 1. 在HBase中创建一个张表
                    checkTable(tp);  //
                    // 2. 把配置放入到广播状态中
                    BroadcastState<String, TableProcess> tpState = ctx.getBroadcastState(tpStateDesc);
                    tpState.put(tp.getSource_table(), tp);
                    
                }
                
                private void checkTable(TableProcess tp) throws SQLException {
                    // jdbc操作的流程
                    // 1. 获取jdbc连接
                    // 2. 定义一个sql语句
                    // create table if not exists user_info(a string, b string, constraint pk primary key(a, b)) SALT_BUCKETS = 3
                    StringBuilder sql = new StringBuilder();  // TODO
                    sql
                        .append("create table if not exists ")
                        .append(tp.getSink_table())
                        .append("(")
                        // id,name,category1_id
                        .append(tp.getSink_columns().replaceAll("([^,]+)", "$1 varchar"))
                        .append(", constraint pk primary key(")
                        .append(tp.getSink_pk() == null ? "id" : tp.getSink_pk())  // "null"
                        .append("))")
                        .append(tp.getSink_extend() == null ? "" :tp.getSink_extend());  // "a"+null -> "anull"
                    
                    
                    System.out.println("建表语句:" + sql.toString());
                    // 3. 通过连接得到一个预处理语句
                    PreparedStatement ps = conn.prepareStatement(sql.toString());
                    // 4. 如果sql中有占位符, 需要给占位符进行赋值.   dml语句
                    // ddl 中没有占位符
                    // 5. 执行预处理语句
                    ps.execute();
                    // 6. 提交
                    conn.commit();
                    // 7.关闭预处理语句
                    ps.close();
                    
                }
            });
           
        
        // 4. 对数据流中的数据进行处理
        
    }
    
    // 读取维度表的配置文件
    private SingleOutputStreamOperator<TableProcess> readTableProcess(StreamExecutionEnvironment env) {
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
            .hostname("hadoop162")
            .port(3306)
            .scanNewlyAddedTableEnabled(true) // eanbel scan the newly added tables fature
            .databaseList("gmall_config") // set captured database
            .tableList("gmall_config.table_process") // set captured tables [product, user, address]
            .username("root")
            .password("aaaaaa")
            .startupOptions(StartupOptions.initial()) // 第一次启动会同步所有数据, 然后使用bin_log监控变化数据
            .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
            .build();
        
        return env
            .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-cdc")
            .map(json -> {
                System.out.println(json);
                
                /*JSONObject after = JSON.parseObject(json).getJSONObject("after");
                return JSON.parseObject(after.toJSONString(), TableProcess.class);*/
                JSONObject obj = JSON.parseObject(json);
                return obj.getObject("after", TableProcess.class);
            });
        
    }
    
    private SingleOutputStreamOperator<String> etlData(DataStreamSource<String> stream) {
        return stream.filter(json -> {
            
            try {
                JSONObject obj = JSON.parseObject(json);
                // ddl语句不要. insert update
                String type = obj.getString("type");
                val data = obj.getJSONObject("data");
                
                return ("insert".equals(type)
                    || "update".equals(type)
                    || "bootstrap-insert".equals(type))
                    && data != null;
            } catch (Exception e) {
                System.out.println("json数据的格式有问题....");
                return false;
            }
        });
    }
}
